package com.world.rickcloudendpoint.stream.consumer;

import com.world.rickcloudendpoint.config.ConsumerConfig;
import com.world.rickcloudendpoint.config.KafkaConsumerConfig;
import com.world.rickcloudendpoint.stream.consumer.kafka.OffsetManager;
import com.world.rickcloudendpoint.stream.filter.Filter;
import com.world.rickcloudendpoint.stream.filter.FilterConfig;
import com.world.rickcloudendpoint.stream.filter.NoOpFilter;
import com.world.rickcloudendpoint.stream.filter.RickCloudFilter;
import lombok.extern.log4j.Log4j2;

import java.util.Map;

/**
 * @description: 消费者生产工厂
 * @author: jlsong
 * @date: 2022/02/28 23:24:19
 * @version: 1.0
 */
@Log4j2
public class RickConsumerFactory {
    private final KafkaConsumerConfig consumerConfig;
    private final Map<String, String> config = System.getenv();

    private final OffsetManager offsetManager = new OffsetManager();

    /**
     * 消费者初始化
     *
     * @param consumerConfig
     */
    public RickConsumerFactory(KafkaConsumerConfig consumerConfig) {
        this.consumerConfig = consumerConfig;
        // 加载kafka队列配置项
        // 初始化指标统计工具
        // 初始化解析器类型
    }

    private RickCloudFilter buildFilter(FilterConfig config) {
        // 打印调用过滤引擎开始
        Filter filter;
        // 通过配置不同的过滤引擎，调用不同的过滤方案
//        switch (config.getFilterEngine()){
//            case "JSON":
//                //
//        }
        filter = new NoOpFilter();
        return new RickCloudFilter(filter);
    }

    public RickConsumer buildConsumer() {
        // 获取过滤配置
        FilterConfig filterConfig = new FilterConfig();
        // 构建过滤器
        RickCloudFilter filter = buildFilter(filterConfig);
        // 创建追踪器
        // 创建消费者
        // 创建水流追踪器

        // 判断kakfa消费模式，如果是SYNC，则使用同步流消费模式
        // 如果不是SYNC，则使用异步流消费模式

        return null;
    }
}
